/**
 * Copyright 2007-2008. Chongqing First Information & Network Co., Ltd. All
 * rights reserved. <a>http://www.cqfirst.com.cn</a>
 */
package com.cqcis.uip.base.framework.handlers.common;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

import com.cqcis.uip.base.framework.FrameworkConstants;
import com.cqcis.uip.base.framework.UIPTaskException;
import com.cqcis.uip.base.framework.common.State;
import com.cqcis.uip.base.framework.common.Task;
import com.cqcis.uip.base.framework.handlers.TableDispatchService;
import com.cqcis.uip.base.framework.utils.ConcurrentTaskQueue;

/**
 * @author huangmy
 * @date   2007-12-17
 */
public abstract class AbstractDispatchHandler extends AbstractHandler 
           implements TableDispatchService {
	private static Log log = LogFactory.getLog(AbstractDispatchHandler.class);	
	private static String LOCK_RESULT = "LOCK_RESULT";
	private ConcurrentTaskQueue<Task> sendTaskQueue;  /*get任务队列*/
	private ConcurrentTaskQueue<Task> sendTaskResultQueue; /*processResult任务队列*/
	private ConcurrentTaskQueue<Task> sendTaskResultPriorityQueue;	
	private ConcurrentTaskQueue<Task> sendTaskResultControlQueue;
	private Map<String, Object> locks = new ConcurrentHashMap<String, Object>();
		
	private void getPriorityResults(List<Task> results) {
		Task priorityTask = null;
		while ((priorityTask = sendTaskResultPriorityQueue.remove()) != null) {
			results.add(priorityTask);
			if (results.size() >= 8192) {					
				return;
			}
		}
	}
	
	private boolean forPriorityResult(List<Task> results) {
		synchronized (this.locks) {	
			if (this.locks.get(LOCK_RESULT) != null) {
				return true;
			}
			
			if (sendTaskResultPriorityQueue.size() <= 0) {
				return false;
			}
			
			getPriorityResults(results);
			
			if (results.size() > 0) {
				this.locks.put(LOCK_RESULT, results);
				return true;
			}
		}
		
		return false;
	}
	
	/**
	 * 获取send的task结果
	 * 
	 * @return true 非正常优先级 , false 正常优先级
	 */
	protected boolean getTaskResult(List<Task> results) {
		sendTaskResultControlQueue.poll();
		
		if (forPriorityResult(results)) {
			return true;
		}
		
		Task result = null;		
		while ((result = sendTaskResultQueue.remove()) != null) {
			results.add(result);
			if (results.size() >= 1024 || sendTaskResultPriorityQueue.size() > 0) {
				break;
			}
		}
		
		return false;
	}
	
	/**
	 * 处理指定的task
	 * 
	 * @param task
	 */
	public void doProcessTaskFor(Task task) {
		boolean opened = false;
		
		try {					
			task.setState(State.SUCCESS);
			opened = schedule(task);			
		} catch (Throwable e) {
			log.error("schedule task Exception: Task=" + task, e);
			task.setThrowable(e);
			opened = false;
			String preMsg = (e instanceof UIPTaskException) ? "" : (FrameworkConstants.TASK_STATE_MSG_INTERFACE_PROCESS_ERROR + ":");
			task.setMessage(preMsg + e.getMessage(), e);
		} finally {			
			if (!opened) {				
				task.setState(State.ERROR);				
			}			
			dispatchTaskResult(task, false);
		}		
	}
			
	/**
	 * 处理需要发送的任务
	 * 
	 * @return
	 */
	private int getProcessTask() {
		List<Task> tasks = null;
		
		try {
			tasks = get(getInterfaceId());
		} catch (Exception e) {
			log.error("get Exception: ", e);	
			return 0;
		}
		
		if (tasks == null) {
			return 0;
		}
		
		for (Task task : tasks) {						
			sendTaskQueue.offer(task);
		}
		
		return tasks.size();
	}
	
	/**
	 * 处理任务结果
	 * 
	 * @return
	 */
	private boolean processTaskResult() {		
		List<Task> results = new ArrayList<Task>();
		boolean normal = getTaskResult(results);		
		if (results != null && results.size() > 0) {			
			boolean s = false;
			try {				
				if (normal) { // true 非正常优先级					
					synchronized (this.locks) {
						try {
						    s = processResult(results);
						} finally {
						    this.locks.remove(LOCK_RESULT);
						}
					}					
				} else {
					s = processResult(results);
				}
			} catch (Throwable e) {
				log.error("Catch Exception in Result: ", e);
			} finally {
				for (Task r : results) {
					getAfterResult().afterProcessResult(r);
				}
			}
			
			return s;
		}
				
		return false;
	}
	
	public Runnable getScheduleRunnable() {
		return new Runnable() {			
			public void run() {
				while (true) {
					Task task = getTask();
					if (task == null) {
						return;
					}
					try {								
						doProcessTaskFor(task);
					} catch(Throwable t) {
						if (t instanceof InterruptedException) {
							return;
						}
						log.error("scheduleTask Exception: ", t);						
					}
				}			 				
	        }
		};
	}
	
	public Runnable getProcessTaskRunnable() {
		return new Runnable() {
			public void run() {
				while (true) {
					try {
						int size = getProcessTask();
				        if (size <= 0) {
				        	break;
				        }
					} catch(Throwable t) {
						log.error("getProcessTask Exception: ", t);
						throw new RuntimeException(t);
					}
				}
			}
		};
	}
	
	public Runnable getProcessTaskResultRunnable() {
		return new Runnable() {
			public void run() {				
				while (true) {
					try {
						boolean success = processTaskResult();
						if (!success) {
							break;
						}						
					} catch (Throwable t) {
						log.error("Process TaskResult action Exception: ", t);
						throw new RuntimeException(t);
					}
				}
			}
		};
	}
	    
    public void registerSendTaskQueue(ConcurrentTaskQueue<Task> sendTaskQueue) {
    	this.sendTaskQueue = sendTaskQueue;
    }
    
    public void registerSendTaskResultQueue(ConcurrentTaskQueue<Task> sendTaskResultQueue) {
    	this.sendTaskResultQueue = sendTaskResultQueue;
    }
    
    public void registerSendTaskResultPriorityQueue(ConcurrentTaskQueue<Task> sendTaskResultPriorityQueue) {
    	this.sendTaskResultPriorityQueue = sendTaskResultPriorityQueue;
    }

	public void registerSendTaskResultControlQueue(ConcurrentTaskQueue<Task> sendTaskResultControlQueue) {
		this.sendTaskResultControlQueue = sendTaskResultControlQueue;
	}
    
}
